package com.smartboot.plus.features.acl;

import com.smartboot.plus.condition.ConditionQ;
import com.smartboot.plus.features.NodeUtil;
import com.smartboot.plus.features.auth.MqttAuthFeatures;
import com.smartboot.plus.holder.BrokerContextHolder;
import com.smartboot.plus.ignite.cache.MqttAclCache;
import com.smartboot.plus.service.MqttAclService;
import com.smartboot.plus.utils.Func;
import org.apache.ignite.Ignite;
import org.smartboot.socket.transport.AioSession;
import tech.smartboot.feat.cloud.annotation.Autowired;
import tech.smartboot.feat.cloud.annotation.Bean;
import tech.smartboot.feat.cloud.annotation.PostConstruct;
import tech.smartboot.feat.core.common.logging.Logger;
import tech.smartboot.feat.core.common.logging.LoggerFactory;
import tech.smartboot.mqtt.common.enums.MqttConnectReturnCode;
import tech.smartboot.mqtt.common.message.MqttMessage;
import tech.smartboot.mqtt.common.message.MqttSubscribeMessage;
import tech.smartboot.mqtt.common.message.MqttVariableHeader;
import tech.smartboot.mqtt.common.message.payload.MqttUnsubscribePayload;
import tech.smartboot.mqtt.common.message.variable.MqttDisconnectVariableHeader;
import tech.smartboot.mqtt.common.message.variable.MqttPubQosVariableHeader;
import tech.smartboot.mqtt.common.message.variable.MqttPublishVariableHeader;
import tech.smartboot.mqtt.common.message.variable.MqttSubscribeVariableHeader;
import tech.smartboot.mqtt.plugin.spec.BrokerContext;
import tech.smartboot.mqtt.plugin.spec.MqttSession;

import java.lang.reflect.Field;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * MQTT ACL（访问控制列表）核心处理类。
 * 实现发布、订阅、取消订阅、断连等操作的权限校验逻辑。
 */
@Bean
public class MqttAclFeatures {

    private static final Logger LOGGER = LoggerFactory.getLogger(MqttAclFeatures.class);

    private static volatile Field subscribersField;

    private static volatile Field mqttUnsubscribePayload;
    // 用于分布式锁防止重复加载 ACL 缓存
    private final ConcurrentMap<String, Object> lockMap = new ConcurrentHashMap<>();

    // ACL 缓存过期时间：5分钟
    private static final long ACL_EXPIRE_MILLIS = 5 * 60 * 1000;

    @Autowired
    private Ignite igniteInstance;


    @Autowired
    private BrokerContext brokerContext;


    @Autowired
    private MqttAclService mqttAclService;

    @Autowired
    private MqttAuthFeatures mqttAuthFeatures;



    private String localNodeId;

    /**
     * 初始化 Ignite 缓存区域
     */
    @PostConstruct
    public void init() {

        localNodeId = NodeUtil.getNodeId(brokerContext);
    }

    /**
     * 处理客户端断开连接
     */
    public boolean disconnectMessage(AioSession session, MqttVariableHeader variableHeader) {

        if (variableHeader instanceof MqttDisconnectVariableHeader) {
            MqttSession mqttSession = session.getAttachment();

          //  Set<String> clientTopics = topicRouter.getTopicsForClient(mqttSession.getClientId());
          //  System.out.println("del="+clientTopics);


            mqttAuthFeatures.getMqttUser().remove(mqttSession.getClientId());
          //  topicRouter.forceCleanupClient(mqttSession.getClientId(),localNodeId);

          // clientTopics = topicRouter.getTopicsForClient(mqttSession.getClientId());
           // System.out.println("del="+clientTopics);
        }


        return true;
    }

    /**
     * 处理取消订阅请求
     */
    public boolean unsubscribeMessage(AioSession session, MqttMessage message, MqttVariableHeader variableHeader) {
        if (variableHeader instanceof MqttPubQosVariableHeader) {
            MqttSession mqttSession = session.getAttachment();
            try {
                Field field = getField(message, "mqttUnsubscribePayload");
                MqttUnsubscribePayload payload = (MqttUnsubscribePayload) field.get(message);  // 读取字段值
                //String nodeId = brokerContext.Options().getNodeId();
                //  String topic = payload.topics().get(0);
                for (String topic : payload.topics()) {
                    // 从路由中移除订阅
                   // topicRouter.removeSubscription(mqttSession.getClientId(), topic, localNodeId);
                    System.out.println("========DUA");
                }
              //  Set<String> clientTopics = topicRouter.getTopicsForClient(mqttSession.getClientId());
            //    System.out.println(clientTopics);
//
            } catch (IllegalAccessException e) {
                throw new RuntimeException(e);
            }

        }
        return true;
    }

    /**
     * 处理发布消息权限校验
     */
    public boolean publishMessage(AioSession session, MqttVariableHeader variableHeader, MqttMessage message) {
        if (variableHeader instanceof MqttPublishVariableHeader) {
            MqttPublishVariableHeader publishVariableHeader = (MqttPublishVariableHeader) variableHeader;
            MqttSession mqttSession = session.getAttachment();

            // 必须是已鉴权用户
            if (!mqttSession.isAuthorized()) return false;
            String clientId = mqttSession.getClientId();
            if (clientId.toUpperCase().startsWith("CLUSTER_HYB_")) {
                return true;
            }

            String username = mqttAuthFeatures.getMqttUser().get(mqttSession.getClientId());
            if (Func.isEmpty(username)) {
                MqttSession.connFailAck(MqttConnectReturnCode.NOT_AUTHORIZED, mqttSession);
                return false;
            }


            // 检查是否有发布权限
            if (!isAllowed(username, publishVariableHeader.getTopicName(), "pub")) {
                MqttSession.connFailAck(MqttConnectReturnCode.NOT_AUTHORIZED, mqttSession);
                return false;
            }


        }
        return true;
    }

    /**
     * 处理订阅消息权限校验
     */
    public boolean subscribeMessage(AioSession session, MqttMessage message, MqttVariableHeader variableHeader) {
        if (variableHeader instanceof MqttSubscribeVariableHeader) {
            MqttSubscribeMessage mqttSubscribeMessage = (MqttSubscribeMessage) message;
            if (mqttSubscribeMessage != null && !mqttSubscribeMessage.getPayload().getTopicSubscriptions().isEmpty()) {
                String topic = mqttSubscribeMessage.getPayload().getTopicSubscriptions().get(0).getTopicFilter();
                MqttSession mqttSession = session.getAttachment();
                // 必须是已鉴权用户
                if (!mqttSession.isAuthorized()) return false;

                String clientId = mqttSession.getClientId();
                if (clientId.toUpperCase().startsWith("CLUSTER_HYB_")) {
                    return true;
                }

                String username = mqttAuthFeatures.getMqttUser().get(clientId);
                if (Func.isEmpty(username)) {
                    MqttSession.connFailAck(MqttConnectReturnCode.NOT_AUTHORIZED, mqttSession);
                    return false;
                }

                // 检查是否有订阅权限
                if (!isAllowed(username, topic, "sub")) {
                    MqttSession.connFailAck(MqttConnectReturnCode.NOT_AUTHORIZED, mqttSession);
                    return false;
                }


            }
        }
        return true;
    }

    private Field getField(MqttMessage mqttSession, String name) {
        try {
//            Field[] declaredFields = mqttSession.getClass().getDeclaredFields();
//            System.out.println(declaredFields);

            if (subscribersField == null) {
                synchronized (MqttAclFeatures.class) {
                    if (subscribersField == null) {
                        // 获取字段并设置为可访问 subscribers
                        subscribersField = mqttSession.getClass().getDeclaredField(name);
                        subscribersField.setAccessible(true);
                    }
                }
            }
            return subscribersField;
        } catch (Exception e) {

        }
        return null;
    }

    private Field getField(MqttSession mqttSession, String name) {
        try {
//            Field[] declaredFields = mqttSession.getClass().getDeclaredFields();
//            System.out.println(declaredFields);

            if (mqttUnsubscribePayload == null) {
                synchronized (MqttAclFeatures.class) {
                    if (mqttUnsubscribePayload == null) {
                        // 获取字段并设置为可访问 subscribers
                        mqttUnsubscribePayload = mqttSession.getClass().getDeclaredField(name);
                        mqttUnsubscribePayload.setAccessible(true);
                    }
                }
            }
            return mqttUnsubscribePayload;
        } catch (Exception e) {

        }
        return null;
    }

    /**
     * 判断是否允许指定操作（发布/订阅）
     */
    public boolean isAllowed(String username, String topic, String action) {
        MqttAclCache cache = getMqttAclCache(username);
        if (cache == null) return false;

        for (MqttAclCache.TopicEft eft : cache.getTopicEft()) {
            if ((eft.getAction().equals(action) || "all".equals(eft.getAction()))
                    && matchTopicPattern(eft.getTopic().toLowerCase(), topic.toLowerCase())
                    && "allow".equalsIgnoreCase(eft.getEffect())) {
                return true;
            }
        }
        return false;
    }

    /**
     * 获取 ACL 缓存
     */
    public MqttAclCache getMqttAclCache(String username) {
        MqttAclCache cacheValue = BrokerContextHolder.getMqttAclCacheMap().get(username);
        if (cacheValue != null) {
            return cacheValue;
        }
        Object lock = lockMap.computeIfAbsent(username, k -> new Object());
        synchronized (lock) {
            try {
                cacheValue = BrokerContextHolder.getMqttAclCacheMap().get(username);
                if (cacheValue != null) {
                    return cacheValue;
                }
                // 从数据库加载 ACL 权限配置
                String sql = new ConditionQ(MqttAclCache::getUsername, username).toString();
                MqttAclCache fresh = mqttAclService.selectOne(sql, MqttAclCache.class);
                if (fresh != null) {
                    //用户编辑时候要清空
                    BrokerContextHolder.getMqttAclCacheMap().put(username, fresh);
                }
                return fresh;
            } finally {
                lockMap.remove(username, lock);
            }
        }
    }

//    public MqttAclCache getMqttAclCache(String username) {
//        CacheValue<MqttAclCache> cacheValue = BrokerContextHolder.getMqttAclCacheMap().get(username);
//        if (cacheValue != null && !cacheValue.isExpired(ACL_EXPIRE_MILLIS)) {
//            return cacheValue.getValue();
//        }
//        Object lock = lockMap.computeIfAbsent(username, k -> new Object());
//        synchronized (lock) {
//            try {
//                cacheValue = BrokerContextHolder.getMqttAclCacheMap().get(username);
//                if (cacheValue != null && !cacheValue.isExpired(ACL_EXPIRE_MILLIS)) {
//                    return cacheValue.getValue();
//                }
//                // 从数据库加载 ACL 权限配置
//                String sql = new ConditionQ(MqttAclCache::getUsername, username).toString();
//                MqttAclCache fresh = mqttAclService.selectOne(sql, MqttAclCache.class);
//                if (fresh != null) {
//                    BrokerContextHolder.getMqttAclCacheMap().put(username, new CacheValue<>(fresh));
//                }
//                return fresh;
//            } finally {
//                lockMap.remove(username, lock);
//            }
//        }
//    }

    /**
     * 主题匹配（支持通配符 +/# $）
     *
     * @param pattern
     * @param topic
     * @return
     */
    private boolean matchTopicPattern(String pattern, String topic) {

        pattern = cleanSharedPrefix(pattern.trim().toLowerCase());
        topic = cleanSharedPrefix(topic.trim().toLowerCase());

        String[] patternParts = pattern.split("/");
        String[] topicParts = topic.split("/");

        int i = 0;
        for (; i < patternParts.length; i++) {
            String p = patternParts[i];

            // '#' 匹配剩余所有层级
            if ("#".equals(p)) {
                return i == patternParts.length - 1;
            }

            if (i >= topicParts.length) {
                return false;
            }

            String t = topicParts[i];
            if (!"+".equals(p) && !p.equals(t)) {
                return false;
            }
        }

        return i == topicParts.length;
    }

    /**
     * 清理共享订阅前缀：如果是 "$share/group/topic" 结构，就返回 "topic"
     */
    private String cleanSharedPrefix(String fullTopic) {
        if (fullTopic.startsWith("$share/")) {
            int secondSlash = fullTopic.indexOf('/', 7); // 7 是 "$share/" 的长度
            if (secondSlash > 0 && secondSlash < fullTopic.length() - 1) {
                return fullTopic.substring(secondSlash + 1);
            } else {
                return ""; // 非法共享结构，返回空防止误匹配
            }
        }
        return fullTopic;
    }

}
